package com.fs.iot.service.flink;

import com.alibaba.fastjson2.JSONObject;
import com.fs.iot.bean.dto.DeviceModelFieldDTO;
import lombok.Data;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Flink动态开窗处理器
 *
 * @author  LiuQi 2025/3/27-7:33
 * @version V1.0
 **/
public class FlinkWindowProcessFunction extends KeyedProcessFunction<String, JSONObject, ProcessResult> {
    private transient MapState<WindowKey, List<String>> windowState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        windowState = getRuntimeContext()
                .getMapState(new MapStateDescriptor<>("windowState", Types.GENERIC(WindowKey.class), Types.LIST(Types.GENERIC(String.class))));
    }

    @Override
    public void processElement(JSONObject obj, KeyedProcessFunction<String, JSONObject, ProcessResult>.Context ctx, Collector<ProcessResult> out) throws Exception {
        long ts = obj.getLong("__ts");
        Map<String, DeviceModelFieldDTO> fieldMap = (Map<String, DeviceModelFieldDTO>) obj.get("__fields");

        // 一个个参数处理
        for (Map.Entry<String, DeviceModelFieldDTO> entry : fieldMap.entrySet()) {
            String key = entry.getKey();
            DeviceModelFieldDTO field = entry.getValue();
            Object value = obj.get(key);
            if (null == value) {
                continue;
            }

            // 如果不用开窗，这边直接使用out进行收集


            // 获取字段开窗参数
            long windowStart = ts - (ts % 5);
            long windowEnd = windowStart + 10;
            WindowKey windowKey = new WindowKey(key, windowStart, windowEnd);
            // 更新窗口状态
            List<String> values = windowState.get(windowKey);
            if (values == null) {
                values = new ArrayList<>();
            }
            values.add(String.valueOf(value));
            windowState.put(windowKey, values);

            // 注册窗口结束的定时器（事件时间）
            ctx.timerService().registerEventTimeTimer(windowEnd);
        }
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, ProcessResult>.OnTimerContext ctx, Collector<ProcessResult> out) throws Exception {
        super.onTimer(timestamp, ctx, out);

        // 遍历所有窗口键，找出结束时间匹配的窗口
//        for (WindowKey key : windowState.keys()) {
//            if (key.getWindowEnd() == timestamp) {
//                List<String> values = windowState.get(key);
//                if (values != null && !values.isEmpty()) {
//                    out.collect(new WindowResult(
//                            key.getParam(),
//                            values,
//                            key.getWindowStart(),
//                            key.getWindowEnd()
//                    ));
//                    windowState.remove(key); // 清理状态
//                }
//            }
//        }
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    // 辅助类：窗口键（参数名 + 窗口时间范围）
    @Data
    private static class WindowKey implements Serializable {
        private final String param;
        private final long windowStart;
        private final long windowEnd;

        public WindowKey(String param, long windowStart, long windowEnd) {
            this.param = param;
            this.windowStart = windowStart;
            this.windowEnd = windowEnd;
        }

//        // 必须实现 hashCode 和 equals
//        @Override
//        public int hashCode() { /* 根据字段计算 */ }
//        @Override
//        public boolean equals(Object obj) { /* 字段相等判断 */ }
    }
}
